#include "cgzCTcpS.h"
#include "cgzCThreadPool.h"

cgzCLoopS::cgzCLoopS(ETCP_S etype)
    : type(etype)
{
}

cgzCLoopS::~cgzCLoopS()
{
}

ETCP_S cgzCLoopS::getEType() const
{
    return type;
}

cgzCAcptS::cgzCAcptS(int fd, int ufd)
    : cgzCLoopS(TCPS_ACPT),
      m_fd(fd),
      m_ufd(ufd)
{
}

cgzCAcptS::~cgzCAcptS()
{
}

cgzCSendS::cgzCSendS(const int ufd)
    : cgzCLoopS(TCPS_SEND),
      m_ufd(ufd),
      m_sendlen(0)
{
}

cgzCSendS::cgzCSendS(cgzHeader &header, const int ufd)
    : cgzCLoopS(TCPS_SEND),
      m_ufd(ufd),
      m_len(header.len + sizeof(cgzGateHeader)),
      m_sendlen(0)
{
    // 重新构造消息头
    m_data = (char *)malloc(m_len);
    cgzGateHeader gateHeader{header.mid, header.len, m_len, header.cid, ufd};
    memcpy(m_data, &gateHeader, sizeof(cgzGateHeader));
    // 拷贝消息体
    auto date = m_data + sizeof(cgzGateHeader);
    memcpy(m_data, date, header.len);
}

cgzCSendS::cgzCSendS(
    const int gfd,
    const int mid,
    const char *data,
    const int len,
    const int cfd,
    const long cid)
    : cgzCLoopS(TCPS_SEND),
      m_ufd(gfd),
      m_len(len + sizeof(cgzGateHeader)),
      m_sendlen(0)
{
    // 拷贝发送数据
    m_data = (char *)malloc(m_len);
    auto &header = *(cgzGateHeader *)m_data;
    header.mid = mid;
    header.len = len;
    header.cid = cid;
    header.all = m_len;
    header.ufd = cfd;
    memcpy(m_data + sizeof(cgzGateHeader), data, len);
}

cgzCSendS::cgzCSendS(
    const int ufd,
    const int mid,
    const int len,
    const long cid)
    : cgzCLoopS(TCPS_SEND),
      m_ufd(ufd),
      m_len(len + sizeof(cgzHeader)),
      m_sendlen(0)

{
    // 拷贝发送数据
    m_data = (char *)malloc(m_len);
    auto &header = *(cgzHeader *)m_data;
    header.mid = mid;
    header.len = len;
    header.cid = cid;
    header.all = m_len;
}

cgzCSendS::cgzCSendS(
    const int ufd,
    const int mid,
    const char *data,
    const int len,
    const long cid)
    : cgzCLoopS(TCPS_SEND),
      m_ufd(ufd),
      m_len(len + sizeof(cgzHeader)),
      m_sendlen(0)

{
    // 拷贝发送数据
    m_data = (char *)malloc(m_len);
    auto &header = *(cgzHeader *)m_data;
    header.mid = mid;
    header.len = len;
    header.cid = cid;
    header.all = m_len;
    memcpy(m_data + sizeof(cgzHeader), data, len);
}

cgzCSendS::~cgzCSendS()
{
    free(m_data);
    m_data = nullptr;
    m_len = 0;
    m_sendlen = 0;
}

cgzCRecvS::cgzCRecvS(const char *data, const int len, const int ufd, const int fd)
    : cgzCLoopS(TCPS_RECV),
      m_data(nullptr),
      m_len(len - sizeof(uint16_t)),
      m_ufd(ufd),
      m_fd(fd)
{
    // 拷贝接收数据
    m_data = (char *)malloc(m_len);
    memcpy(m_data, data + sizeof(uint16_t), m_len);
}

cgzCRecvS::cgzCRecvS(cgzHeader &header, const int ufd, const int fd)
    : cgzCLoopS(TCPS_RECV),
      m_len(header.all),
      m_ufd(ufd),
      m_fd(fd)
{
    // 拷贝接收数据
    m_data = (char *)malloc(m_len);
}

cgzCRecvS::cgzCRecvS(cgzGateHeader &header, const int ufd, const int fd)
    : cgzCLoopS(TCPS_RECV),
      m_len(header.all),
      m_ufd(ufd),
      m_fd(fd)
{
    // 拷贝接收数据
    m_data = (char *)malloc(m_len);
}

cgzCRecvS::~cgzCRecvS()
{
    free(m_data);
}

cgzCDiscS::cgzCDiscS(int ufd, int fd)
    : cgzCLoopS(TCPS_DISC),
      m_ufd(ufd),
      m_fd(fd)
{
}

cgzCDiscS::~cgzCDiscS()
{
}

cgzCGAddS::cgzCGAddS()
    : cgzCLoopS(TCPS_GADD)
{
}

cgzCGAddS::~cgzCGAddS()
{
}

cgzCGDelS::cgzCGDelS()
    : cgzCLoopS(TCPS_GDEL)
{
}

cgzCGDelS::~cgzCGDelS()
{
}

cgzCGClrS::cgzCGClrS()
    : cgzCLoopS(TCPS_GCLR)
{
}

cgzCGClrS::~cgzCGClrS()
{
}

cgzCTcpFd::cgzCTcpFd(int fd, int ufd)
    : m_hand(true),
      m_fd(fd),
      m_ufd(ufd),
      m_recv(READLEN)
{
    m_hand = true;
}

cgzCTcpFd::~cgzCTcpFd()
{
}

cgzCTcpS::cgzCTcpS(cgzCThreadPool *pool, const char *name, const std::string ip, const int port, const int cid)
    : cgzITask(pool, name, cgzETask::CTcpS),
      m_ip(ip),
      m_port(port),
      m_cid(cid),
      m_sfd(-1),
      m_efd(-1),
      m_ufd(0)
{
}

cgzCTcpS::~cgzCTcpS()
{
    if (m_efd >= 0)
    {
        for (auto it : m_fdmap)
        {
            closeFd(it.first);
            auto fdinfo = it.second;
            fdinfo->~cgzCTcpFd();
            free(fdinfo);
        }
        if (m_sfd >= 0)
        {
            closeFd(m_sfd);
            m_sfd = -1;
        }
        m_fdmap.clear();
        m_ufdmap.clear();
    }
}

bool cgzCTcpS::closeFd(int fd)
{
    if (m_sfd > 0)
    {
        epoll_ctl(m_efd, EPOLL_CTL_DEL, fd, nullptr);
    }
    if (close(fd) == -1)
    {
        perror("close");
    }
    auto it = m_fdmap.find(fd);
    m_fdmap.erase(it);
    return true;
}

bool cgzCTcpS::run(cgzCLoopS *task)
{
    assert(false);
    return false;
}

bool cgzCTcpS::run(int64_t nms)
{
    while (!m_runTask.empty())
    {
        auto task = m_runTask.front();
        m_runTask.pop_front();
        auto loop = static_cast<cgzCLoopS *>(task);

        if (loop->type == TCPS_SEND)
        {
            auto csend = static_cast<cgzCSendS *>(loop);
            m_sendqueue.push(csend);
        }
        else
        {
            run(loop);
        }
    }

    tcpsend();
    return true;
}

bool cgzCTcpS::tcpbind()
{
    m_sfd = socket(AF_INET, SOCK_STREAM, 0);
    if (m_sfd < 0)
    {
        cgz_this_error("Create socket failed name:{} err:{}", m_name.c_str(), strerror(errno));
        return false;
    }

    struct sockaddr_in addr;
    memset(&addr, 0, sizeof(addr));
    addr.sin_family = AF_INET;
    addr.sin_port = htons(m_port);
    addr.sin_addr.s_addr = inet_addr(m_ip.c_str());

    int ret = bind(m_sfd, (struct sockaddr *)&addr, sizeof(addr));
    if (ret < 0)
    {
        cgz_this_error("Bind socket failed name:{} err:{}", m_name.c_str(), strerror(errno));
        close(m_sfd); // 关闭套接字
        return false;
    }

    // 设置套接字为非阻塞模式
    int flags = fcntl(m_sfd, F_GETFL, 0);
    if (flags == -1)
    {
        cgz_this_error("Failed to get socket flags name:{} err:{}", m_name.c_str(), strerror(errno));
        close(m_sfd);
        return false;
    }
    if (fcntl(m_sfd, F_SETFL, flags | O_NONBLOCK) == -1)
    {
        cgz_this_error("Failed to set socket to non-blocking name:{} err:{}", m_name.c_str(), strerror(errno));
        close(m_sfd);
        return false;
    }

    ret = listen(m_sfd, 10);
    if (ret < 0)
    {
        cgz_this_error("Listen socket failed name:{} err:{}", m_name.c_str(), strerror(errno));
        close(m_sfd);
        return false;
    }

    // 创建epoll句柄
    m_efd = epoll_create1(0);
    struct epoll_event event;
    event.events = EPOLLIN;
    event.data.fd = m_sfd; // 监视监听套接字
    epoll_ctl(m_efd, EPOLL_CTL_ADD, m_sfd, &event);

    // 创建事件线程
    m_pool->addTask(m_name.c_str(), this);
    m_thread = new std::thread(&cgzCTcpS::tcpread, this);
    m_pool->push_thread(m_thread);
    cgz_this_error("Listen on:{} m_port:{} m_sfd:{} m_efd:{}", m_ip.c_str(), m_port, m_sfd, m_efd);
    return true;
}

bool cgzCTcpS::taccpet()
{
    // 接受连接
    struct sockaddr_in client_addr;
    socklen_t client_len = sizeof(client_addr);
    int fd = accept(m_sfd, (struct sockaddr *)&client_addr, &client_len);
    if (fd < 0)
    {
        if (!(errno == EAGAIN || errno == EWOULDBLOCK))
        {
            cgz_this_error("Accept failed name:{} err:{}", m_name.c_str(), strerror(errno));
            return false;
        }
    }
    else
    {
        // taccpet() 接受连接后设置：
        int yes = 1;
        setsockopt(fd, IPPROTO_TCP, TCP_NODELAY, &yes, sizeof(yes)); // 禁用Nagle
        setsockopt(fd, SOL_SOCKET, SO_KEEPALIVE, &yes, sizeof(yes)); // 启用心跳
        // 将新的连接套接字添加到 epoll 实例中
        struct epoll_event client_event;
        client_event.events = EPOLLIN | EPOLLET | EPOLLRDHUP; // 监视可读和挂起事件
        client_event.data.fd = fd;                            // 存储新的连接套接字
        epoll_ctl(m_efd, EPOLL_CTL_ADD, fd, &client_event);

        // 设置套接字为非阻塞模式
        int flags = fcntl(fd, F_GETFL, 0);
        if (flags == -1)
        {
            cgz_this_error("Failed to get socket flags name:{} err:{}", m_name.c_str(), strerror(errno));
            close(fd);
            return false;
        }
        if (fcntl(fd, F_SETFL, flags | O_NONBLOCK) == -1)
        {
            cgz_this_error("Failed to set socket to non-blocking name:{} err:{}", m_name.c_str(), strerror(errno));
            close(fd);
            return false;
        }
        this->connect(fd);
        cgz_this_error("New connection from {}:{} fd:{}", inet_ntoa(client_addr.sin_addr), ntohs(client_addr.sin_port), fd);
    }

    return true;
}

bool cgzCTcpS::todoret(cgzIData *ctodo)
{
    m_synTask.push_back(ctodo);
    return true;
}

bool cgzCTcpS::retTask(cgzIData *cgzTask)
{
    return todoret(cgzTask);
}

bool cgzCTcpS::casBusy()
{
    // 处理返回事件
    std::unique_lock lock(m_rwmutex);
    if (m_synTask.empty())
    {
        return false;
    }
    m_retTask.splice(m_retTask.end(), m_synTask);
    setFree();
    return true;
}

bool cgzCTcpS::tcpsend()
{
    while (!m_sendqueue.empty())
    {
        auto csend = m_sendqueue.front();
        auto it = m_ufdmap.find(csend->m_ufd);
        if (it == m_ufdmap.end())
        {
            // 已经失效的客户端-删除发送请求
            m_sendqueue.pop();
            csend->~cgzCSendS();
            free(csend);
            continue;
        }

        auto fdinfo = it->second;

        while (csend->m_sendlen < csend->m_len)
        {
            auto leftlen = csend->m_len - csend->m_sendlen;
            auto leftptr = csend->m_data + csend->m_sendlen;
            ssize_t sendlen = send(fdinfo->m_fd, leftptr, leftlen, 0);
            if (sendlen < 0)
            {
                if (errno == EAGAIN || errno == EWOULDBLOCK)
                {
                    cgz_this_error("Send would block, try again later name:{} err:{}", m_name.c_str(), strerror(errno));
                    break;
                }
                else
                {
                    cgz_this_error("Send failed name:{} err:{}", m_name.c_str(), strerror(errno));
                    csend->m_sendlen = 0; // 重置发送长度
                    m_ufdmap.erase(it);
                    disstcp(fdinfo);
                    break;
                }
            }
            else
            {
                csend->m_sendlen += sendlen;
            }
        }

        if (csend->m_sendlen >= csend->m_len)
        {
            if (!fdinfo->m_hand)
            {
                handshake(fdinfo);
            }
            m_sendqueue.pop();
            csend->~cgzCSendS();
            free(csend);
        }
        else
        {
            // 无法发送了
            break;
        }
    }
    return true;
}

bool cgzCTcpS::tcpread()
{
    while (m_bRunning)
    {
        struct epoll_event events[EPOOLEC];
        int epoolC = epoll_wait(m_efd, events, EPOOLEC, -1);
        if (epoolC < 0)
        {
            cgz_this_error("Epoll error name:{} err:{}", m_name.c_str(), strerror(errno));
            if (errno != EINTR)
            {
                return false;
            }
        }
        else if (epoolC == 0)
        {
            continue;
        }
        else
        {
            std::unique_lock lock(m_rwmutex);
            for (int i = 0; i < epoolC; ++i)
            {
                if (events[i].data.fd == m_sfd)
                {
                    // 监听套接字可读，接受连接
                    taccpet();
                }
                else
                {
                    auto it = m_fdmap.find(events[i].data.fd);
                    auto fdinfo = it->second;
                    while (true)
                    {
                        auto dire = fdinfo->m_recv.dire();
                        if (dire <= 0)
                        {
                            // 没有接受缓存了区|逻辑错误(非逻辑错误请扩展缓存区)
                            cgz_this_error("接收缓存区可能不够 name:{} dire:{}", m_name, dire);
                            break;
                        }
                        auto leftptr = fdinfo->m_recv.head();
                        auto recvlen = recv(fdinfo->m_fd, leftptr, dire, 0);
                        if (recvlen < 0)
                        {
                            if (errno == EAGAIN || errno == EWOULDBLOCK)
                            {
                                // 读取完成退出
                                break;
                            }
                            else
                            {
                                cgz_this_error("Recv failed name:{} err:{}", m_name, strerror(errno));
                                it = m_ufdmap.find(fdinfo->m_ufd);
                                m_ufdmap.erase(it);
                                disstcp(fdinfo);
                                break;
                            }
                        }
                        else if (recvlen == 0)
                        {
                            // 关闭连接
                            cgz_this_error("closed by peer. name:{} fd: {} ufd: {}", m_name, fdinfo->m_fd, fdinfo->m_ufd);
                            it = m_ufdmap.find(fdinfo->m_ufd);
                            m_ufdmap.erase(it);
                            disstcp(fdinfo);
                            break;
                        }
                        else
                        {
                            fdinfo->m_recv.push(recvlen);
                        }
                    }

                    while (sticking(fdinfo))
                        ;
                }
            }
        }
    }

    return true;
}

bool cgzCTcpS::sticking(cgzCTcpFd *fdinfo)
{
    cgzCireBuffer &recv = fdinfo->m_recv;
    int used = recv.used();

    // 检查消息头部
    if (used < sizeof(cgzHeader))
    {
        return false;
    }

    // 检查消息完整
    cgzHeader header;
    recv.read(&header, sizeof(cgzHeader));
    if (used < header.all)
    {
        return false;
    }

    void *node = malloc(sizeof(cgzCRecvS));
    auto pack = new (node) cgzCRecvS(header, fdinfo->m_ufd, fdinfo->m_fd);
    recv.read(pack->m_data, header.all);
    recv.pop(header.all);

    pack->target = m_cid;
    pack->origin = getId();
    todoret(pack);
    return true;
}

bool cgzCTcpS::connect(int fd, bool is_hand)
{
    int ufd = ++m_ufd;
    void *node = malloc(sizeof(cgzCTcpFd));
    auto *fdinfo = new (node) cgzCTcpFd(fd, ufd);
    m_ufdmap.insert(std::make_pair(ufd, fdinfo));
    m_fdmap.insert(std::make_pair(fd, fdinfo));
    if (is_hand)
    {
        sendaccept(fdinfo);
    }
    fdinfo->m_hand = is_hand;
    return true;
}

bool cgzCTcpS::handshake(cgzCTcpFd *fdinfo)
{
    fdinfo->m_hand = true;
    return true;
}

bool cgzCTcpS::sendaccept(cgzCTcpFd *fdinfo)
{
    void *node = malloc(sizeof(cgzCAcptS));
    auto acpt = new (node) cgzCAcptS(fdinfo->m_fd, fdinfo->m_ufd);
    acpt->target = m_cid;
    acpt->origin = getId();
    todoret(acpt);
    return true;
}

bool cgzCTcpS::disstcp(cgzCTcpFd *fdinfo)
{
    void *node = malloc(sizeof(cgzCDiscS));
    auto disc = new (node) cgzCDiscS(fdinfo->m_ufd, fdinfo->m_fd);
    cgzCThreadPool *pool = cgzCThreadPool::Instance();
    pool->addTask(getName(), "cgzCLua", disc);
    closeFd(fdinfo->m_fd);
    close(fdinfo->m_fd);
    fdinfo->~cgzCTcpFd();
    free(fdinfo);
    return true;
}